package com.gx.mvp.rx;

import io.reactivex.Observable;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;

/**
 * 用于管理RxBus的事件和Rxjava相关代码的生命周期处理
 *
 * Created by Guoxin on 2019/12/05.
 */
public class RxManager implements Serializable {

    /**
     * 管理被观察者
     */
    private Map<String, Observable<Object>> mObservableMap = new HashMap<>();

    /**
     * 管理订阅
     */
    private CompositeDisposable mCompositeDisposable = new CompositeDisposable();

    /**
     * 发送事件
     *
     * @param tag
     * @param content
     */
    public void post(Object tag, Object content) {
        RxBus.getInstance().post(tag, content);
    }

    /**
     * 接收事件
     *
     * @param eventName
     * @param consumer
     */
    public void on(String eventName, final Consumer<Object> consumer) {
        final Observable<Object> observable = RxBus.getInstance().register(eventName);
        this.mObservableMap.put(eventName, observable);
        final Disposable disposable = observable.flatMap((Function<Object, Observable<Object>>) o -> {
            try {
                return Observable.just(o);
            } catch (Throwable t) {
                return Observable.error(t);
            }
        }).observeOn(AndroidSchedulers.mainThread()).subscribe(consumer, throwable -> {
            throwable.printStackTrace();
            // 事件失败后再次订阅才能再次接收到事件
            observable.observeOn(AndroidSchedulers.mainThread()).subscribe(consumer);
        });
        this.add(disposable);
    }

    /**
     * 判断事件是否已注册
     *
     * @param eventName
     *
     * @return
     */
    public boolean registered(String eventName) {
        return RxBus.getInstance().registered(eventName);
    }

    /**
     * 添加订阅
     *
     * @param disposable
     */
    public void add(Disposable disposable) {
        this.mCompositeDisposable.add(disposable);
    }

    /**
     * 取消订阅，并清除所有的事件监听
     */
    public void clear() {
        this.mCompositeDisposable.clear();
        this.mCompositeDisposable.dispose();  // 取消订阅
        for (Map.Entry<String, Observable<Object>> entry : this.mObservableMap.entrySet()) {
            RxBus.getInstance().unregister(entry.getKey(), entry.getValue());   // 移除观察
        }
    }
}
